package com.wchen.project.zk.curator;

import com.google.common.collect.Lists;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.List;

/**
 * @Desc:   LeaderLatch 模拟 10 个客户端抢主的情况，客户端称为 leader 之后手动调用 close() 释放 leader 权限并退出 leader 争夺
 *          使用场景：分布式调度系统中，选出一个 Master 集群中的 Leader 来执行任务池中的任务并执行，这样才能避免单点问题与
 *          重复执行问题；
 *
 *          LeaderLatch 基本原理：
 *              选择一个根路径，例如“/leader_select", 多个机器同时向该根路径下创建临时顺序节点，如"/leader_latch/node-3"
 *          "/leader_latch/node_1","/leader_latch/node_2",节点编号最小的zk客户端将成为 leader ，没抢到 Leader 的节点都
 *          监听前一个节点的删除时间，在前一个节点删除后重新抢主；
 *
 *          核心流程就是
 *          1.zk客户端往同一路径下创建临时节点，创建后回调callBack
 *          2.在回调事件中判断自身节点是否是节点编号最小的一个
 *          3.如果是，则抢主成功，如果不是，设置对前一个节点(编号更小的)的删除事件的监听器，删除事件触发后重新进行抢主
 *
 * @author: cww
 * @DateTime: 2020-01-25 16:48
 */

public class LeaderLatchTest {
    static int CLIENT_COUNT = 10;
    static String LOCK_PATH = "/leader_latch";

    public static void main(String[] args) throws Exception {
//        List<CuratorFramework> clientList = Lists.newArrayListWithCapacity(CLIENT_COUNT);
        List<LeaderLatch> leaderLatchList = Lists.newArrayListWithCapacity(CLIENT_COUNT);
        // 创建10 个 zk 客户端模拟 leader 选举
        for (int i = 0; i < CLIENT_COUNT; i++) {
            CuratorFramework client = getClient();
//            clientList.add(client);
            LeaderLatch leaderLatch = new LeaderLatch(client, LOCK_PATH, "CLIENT_" + i);
            leaderLatchList.add(leaderLatch);
            // 必须用 start() 方法进行抢主
            leaderLatch.start();
        }
        // 判断当前 leader 是哪个客户端
        checkLeader(leaderLatchList);

    }

    private static void checkLeader(List<LeaderLatch> leaderLatchList) throws Exception {
        // Leader 选举需要时间 等待
        Thread.sleep(3000);
        for (int i = 0; i < leaderLatchList.size(); i++) {
            LeaderLatch leaderLatch = leaderLatchList.get(i);
            // 通过 hasLeadership() 方法判断当前节点是否是 leader
            if (leaderLatch.hasLeadership()) {
                System.out.println("当前 leader: " + leaderLatch.getId());
                // 释放 leader 权限
                leaderLatch.close();
                // 重新进行抢主
                checkLeader(leaderLatchList);
            }
        }
    }

    private static CuratorFramework getClient() {
        String zkServerAddress = "home:2181,home:2182,home:2183";
        CuratorFramework zkClient = CuratorFrameworkFactory.builder()
                .connectString(zkServerAddress)
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(5000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3, 5000))
                .build();
        zkClient.start();
        return zkClient;
    }
}
